-- Quotes Pushlish System with redis
--[[
  Copyright (C) Fei Zi (kuangye)
  created by 2016年8月6日13:34:24
  version 1.0

  default message hava two types from redis
  public message  
  0|data
  the type message push all user.
  private message
  1|key|data
  the type message push the key user.
]]
local server = require "resty.websocket.server"
local redis  = require "resty.redis"
local http   = require "resty.http"
local status = require "status"

local curmsgid = status.get(ngx.worker.pid()) or 0;
local pubchannel = "pubchannel";


function split(str, pat)
   local t = {}  -- NOTE: use {n = 0} in Lua-5.0
   local fpat = "(.-)" .. pat
   local last_end = 1
   local s, e, cap = str:find(fpat, 1)
   while s do
      if s ~= 1 or cap ~= "" then
   table.insert(t,cap)
      end
      last_end = e+1
      s, e, cap = str:find(fpat, last_end)
   end
   if last_end <= #str then
      cap = str:sub(last_end)
      table.insert(t, cap)
   end
   return t
end


function userAuth(token,customerno )
  if token == nil or token == nil then
     return false
  end
  
  if token == '111111' then
     return true
  end
end

local token = ngx.var.arg_token
local customerno = ngx.var.arg_customerno


if not userAuth(token,customerno) then
  ngx.exit(444)
end


--拉取消息
local pull = function()

    local msgid = 0;
    local redisisexception = false;

    local red = redis:new()
    red:set_timeout(5000) -- 5 sec
    local ok, err = red:connect("127.0.0.1", 6379)
    if not ok then
        ngx.log(ngx.ERR, "failed to connect redis: ", err)
        return
    end

    local res, err = red:subscribe(pubchannel)
    if not res then
        ngx.log(ngx.ERR, "failed to subscribe redis: ", err)
        return
    end

    while true do
    
      redisisexception = false;
      local res, err = red:read_reply()
      if res then
            local item = res[3]
            --ngx.log(ngx.ERR, "read_reply response message: ", item)
            ngx.shared.msgqueue:set(msgid,item)
            status.set(ngx.worker.pid(),msgid)  --保存当前生产者msgid
            msgid = msgid + 1
      else
          if err ~= "timeout" then
            --ngx.log(ngx.ERR, "read_reply response error: ", err)
            redisisexception = true
          end
      end
       
      if  redisisexception then
          red = redis:new()
          red:set_timeout(5000) -- 5 sec
          local ok, err = red:connect("127.0.0.1", 6379)
          if not ok then
              ngx.log(ngx.ERR, "failed to reconnect redis: ", err)
              return
          end

          local res, err = red:subscribe(pubchannel)
          if not res then
              redisisexception = true;
              ngx.log(ngx.ERR, "failed to resubscribe redis: ", err)
          end
      end
    end
end

local  version = ngx.config.nginx_version
if  status.get(version) == nil then
    ngx.thread.spawn(pull)
    status.set(ngx.worker.pid(),0)
    status.set(version,0)
    ngx.log(ngx.ERR, "pull module init successfully!")
end

--create connection
local wb, err = server:new{
  timeout = 10000,
  max_payload_len = 65535
}

--create success
if not wb then
  ngx.log(ngx.ERR, "failed to new websocket: ", err)
  return ngx.exit(444)
end


--push public quote info 
local push = function()
    
    -- loop : read from redis
    local res = nil

    while true do
        ngx.sleep(0.005)
        res = ngx.shared.msgqueue:get(curmsgid)
        local continue = true; 

        if res then
          
            local result = split(res,"|");
            local message = nil

            if not result then
                ngx.log(ngx.ERR, "data analysis error: ",item)
                continue = false
            end

            if result[1] == "0" then
                  message = result[2]
            elseif result[1] == "1" then
                 if result[2] == customerno then
                      message = result[3]
                 else
                      continue = false
                 end
            end
            
            if continue then
              local bytes, err = wb:send_text(message)
              if not bytes then
                  -- better error handling
                  ngx.log(ngx.ERR, "failed to send text: ", err)
                  return ngx.exit(444)
              end
            end

            curmsgid = curmsgid + 1   -- 只要有消息进来，就+1
        end
    end
end


local co = ngx.thread.spawn(push)

--main loop
while true do
    -- get data
    local data, typ, err = wb:recv_frame()

    -- if the connection is broken , exit
    if wb.fatal then
        ngx.log(ngx.ERR, "failed to receive frame: ", err)
        return ngx.exit(444)
    end

    if not data then
        local bytes, err = wb:send_ping()
        if not bytes then
          ngx.log(ngx.ERR, "failed to send ping: ", err)
          return ngx.exit(444)
        end
        ngx.log(ngx.ERR, "send ping: ", data)
    elseif typ == "close" then
        break
    elseif typ == "ping" then
        local bytes, err = wb:send_pong()
        if not bytes then
            ngx.log(ngx.ERR, "failed to send pong: ", err)
            return ngx.exit(444)
        end
    elseif typ == "pong" then
        ngx.log(ngx.ERR, "client ponged")
    elseif typ == "text" then

          ngx.log(ngx.ERR, "data from web: ", data)
         
          local httpc = http.new()
        -- The generic form gives us more control. We must connect manually.
          httpc:set_timeout(500)
          httpc:connect("127.0.0.1", 81)

          local res, err = httpc:request({
            path = "/mqu/KjaxAjaxServlet.as",
            method = "POST",
            body = data,
            headers = {
              ["Content-Type"] = "application/x-www-form-urlencoded",
            }
          })

         if err then
            ngx.log(ngx.ERR,"127.0.0.1:81 -> HTTP_SERVICE_UNAVAILABLE")
            wb:send_text("Quotes server request failed,please retry..")
            break
         end

         if not res then
            ngx.log(ngx.ERR,"127.0.0.1:81 -> HTTP_SERVICE_UNAVAILABLE")
            wb:send_text("Quotes server request success,but the server not send data, please retry..")
            break
         end

         local reader = res.body_reader

         --default load 8K data 
         repeat
            local chunk, err = reader(8192)
 
              if err then
                  ngx.log(ngx.ERR, err)
              break
            end

            if chunk then
                wb:send_text(chunk)
            end
        until not chunk

        local ok, err = httpc:set_keepalive()
          if not ok then
             ngx.log("failed to set keepalive: ", err)
        end
    end
end

wb:send_close()
ngx.thread.wait(co)